package com.yunlongn.async.log.impl;


import com.yunlongn.async.config.RetrySchedulerConfig;
import com.yunlongn.async.log.LogDo;
import com.yunlongn.common.json.JsonUtils;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Optional;
import java.util.Set;

/**
 * @author yunlong.li
 */
public class RedisScheduleServiceImpl extends AbstractScheduleLog {

    private final static String CACHE_KEY = "retry_schedule_log";
    private final static String CACHE_KEY_SCOPE = "retry_schedule_scope";

    private static final StringRedisTemplate STRING_REDIS_TEMPLATE = RetrySchedulerConfig.getStringRedisTemplate();

    @Override
    protected void doSava(LogDo logDo) {
        doDelete(logDo.getId());
        STRING_REDIS_TEMPLATE.opsForZSet().add(CACHE_KEY_SCOPE, logDo.getId(), logDo.getNextRetryTime());
        STRING_REDIS_TEMPLATE.opsForHash().put(CACHE_KEY, logDo.getId(), JsonUtils.toJsonString(logDo));
    }

    @Override
    protected LogDo doQuery(String id) {
        Object doQueryObj = STRING_REDIS_TEMPLATE.opsForHash().get(CACHE_KEY, id);
        if (!ObjectUtils.isEmpty(doQueryObj)) {
            return JsonUtils.fromJson((String) doQueryObj, LogDo.class);
        }
        return null;
    }

    @Override
    protected void doUpdate(LogDo entity) {
        if (LogDo.Stat.SUCCESS.getStat().equalsIgnoreCase(entity.getStatus())
                || LogDo.Stat.FAIL.getStat().equalsIgnoreCase(entity.getStatus())) {
            doDelete(entity.getId());
        } else {
            doSava(entity);
        }
    }

    private void doDelete(String id) {
        STRING_REDIS_TEMPLATE.opsForZSet().remove(CACHE_KEY_SCOPE, id);
        STRING_REDIS_TEMPLATE.opsForHash().delete(CACHE_KEY, id);
    }

    @Override
    public Iterator<LogDo> loadNotFinishTask() {

        return new Iterator<LogDo>() {

            private int offset = 0;
            private LogDo nextLog = null;

            @Override
            public boolean hasNext() {
                // 从排序中获取一个数据
                Set<String> result = STRING_REDIS_TEMPLATE.opsForZSet().rangeByScore(
                        CACHE_KEY_SCOPE,
                        // 0 ～ 当前时间
                        0d, System.currentTimeMillis() ,
                        // (offset,1)
                        offset, 1);
                if (ObjectUtils.isEmpty(result)) {
                    return false;
                }
                for (String id : result) {
                    String valueStr = (String) STRING_REDIS_TEMPLATE.opsForHash().get(CACHE_KEY, id);
                    if (!StringUtils.hasText(valueStr)) {
                        STRING_REDIS_TEMPLATE.opsForZSet().remove(CACHE_KEY_SCOPE, id);
                        return false;
                    }
                    nextLog = JsonUtils.fromJson(valueStr, LogDo.class);
                    if (nextLog == null) {
                        return false;
                    }

                    // 获取为等待状态的数据
                    return LogDo.Stat.PENDING.getStat().equalsIgnoreCase(nextLog.getStatus());
                }
                return false;
            }

            @Override
            public LogDo next() {
                offset++;
                return nextLog;
            }

            @Override
            public void remove() {
                STRING_REDIS_TEMPLATE.opsForZSet().remove(CACHE_KEY_SCOPE, nextLog.getId());
                STRING_REDIS_TEMPLATE.opsForHash().delete(CACHE_KEY, nextLog.getId());
            }

        };

    }
}
